package com.kitten.note.biz.listener;

import com.kitten.framework.common.utils.JsonUtils;
import com.kitten.note.biz.constants.MQConstants;
import com.kitten.note.biz.constants.RedisKeyConstants;
import com.kitten.note.biz.convert.NoteConvert;
import com.kitten.note.biz.domain.dataobject.NoteDO;
import com.kitten.note.biz.domain.mapper.NoteDOMapper;
import com.kitten.note.biz.enums.NoteOperateEnum;
import com.kitten.note.biz.model.dto.NoteOperateMqDTO;
import com.kitten.note.biz.model.dto.PublishNoteDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;

/**
 * @author kitten
 */
@Slf4j
@RocketMQTransactionListener
public class PublishNote2DBLocalTransactionListener implements RocketMQLocalTransactionListener {

    @Resource
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Resource
    private NoteDOMapper noteDOMapper;

    /**
     * 事务消息执行本地事务
     * @param msg
     * @param arg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 1. 解析消息内容
        String payload = new String((byte[]) msg.getPayload());
        log.info("======> 事务消息: 开始执行本地事务：{}", payload);

        PublishNoteDTO publishNoteDTO = JsonUtils.parseObject(payload, PublishNoteDTO.class);

        Long noteId = publishNoteDTO.getId();
        Long creatorId = publishNoteDTO.getCreatorId();

        // 删除个人主页: 发布笔记列表Redis缓存 note:published:list:userId
        String publishedNoteListKey = RedisKeyConstants.buildPublishedNoteListKey(creatorId);
        redisTemplate.delete(publishedNoteListKey);

        // 2. 本地事务
        try {
            // DTO 转 DO
            NoteDO noteDO = NoteConvert.INSTANCE.convertDTO2DO(publishNoteDTO);

            // 笔记元数据写库
            noteDOMapper.insert(noteDO);
        } catch (Exception e) {
            log.error("## 笔记元数据存储失败: ", e);
            return RocketMQLocalTransactionState.ROLLBACK; // 回滚事务消息
        }

        // 延迟双删
        sendDelayDeleteRedisPublishedNoteListCacheMQ(creatorId);

        //
        NoteOperateMqDTO noteOperateMqDTO = NoteOperateMqDTO.builder()
                .creatorId(creatorId)
                .noteId(noteId)
                .type(NoteOperateEnum.PUBLISH.getCode()) // 发布笔记
                .build();
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(noteOperateMqDTO)).build();
        String destination = MQConstants.TOPIC_NOTE_OPERATE + ":" + MQConstants.TAG_NOTE_PUBLISH;
        rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 发布笔记 MQ 发送成功, {}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.error("==> 发布笔记 MQ 发送失败, {}", e);
            }
        });


        // TODO 3. 返回事务状态
        return RocketMQLocalTransactionState.COMMIT;
    }

    private void sendDelayDeleteRedisPublishedNoteListCacheMQ(Long userId) {
        Message<String> message = MessageBuilder.withPayload(String.valueOf(userId))
                .build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELAY_DELETE_PUBLISHED_NOTE_LIST_REDIS_CACHE, message,
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        log.info("## 延时删除 Redis 已发布笔记列表缓存消息发送成功...");
                    }

                    @Override
                    public void onException(Throwable e) {
                        log.error("## 延时删除 Redis 已发布笔记列表缓存消息发送失败...", e);
                    }
                },
                3000, // 超时时间
                1 // 延迟级别，1 表示延时 1s
        );
    }

    /**
     * 事务状态回查
     * @param msg
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 1. 解析消息内容
        String payload = new String((byte[]) msg.getPayload());
        log.info("======> 事务消息: 开始事务回查：{}", payload);

        // 2. 检查本地事务状态
        PublishNoteDTO publishNoteDTO = JsonUtils.parseObject(payload, PublishNoteDTO.class);

        Long noteId = publishNoteDTO.getId();
        int count = noteDOMapper.selectCountByNoteId(noteId);


        // 3. 返回最终状态
        return count == 1 ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
    }


}
